package org.example.day20241221;

import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.connector.base.DeliveryGuarantee;
import org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema;
import org.apache.flink.connector.kafka.sink.KafkaSink;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @ClassName: FlinkKafkaDemo01
 * @Author: 孙磊
 * @Date: 2024/12/21 21:36
 * @Since: JDK1.8
 * @Version: 1.0
 */
public class FlinkKafkaDemo01 {
    private final static String kafka_server = "localhost:9092";
    private final static String pub_topic = "sunlei";
    private final static String sub_topic = "sub_sunlei";
    private final static String groupId = "kafka-demo";

    public static void main(String[] args) throws Exception {
        //设置执行环境
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //指定并行度
        env.setParallelism(1);
        //构建kafkaSource
        KafkaSource<String> kafkaSource = KafkaSource.<String>builder()
                .setBootstrapServers(kafka_server) //指定kafka服务
                .setTopics(pub_topic)    //指定topic
                .setGroupId(groupId)   //指定groupID
                .setStartingOffsets(OffsetsInitializer.latest())    //指定offset的位置
                .setValueOnlyDeserializer(new SimpleStringSchema())     //指定反序列化器
                .build();
        DataStreamSource<String> sou = env.fromSource(kafkaSource, //指定数据源
                WatermarkStrategy.noWatermarks(), //指定水位线
                "flink kafka source");
        SingleOutputStreamOperator<String> map = sou.map(new MapFunction<String, String>() {
            @Override
            public String map(String s) throws Exception {
                JSONObject jsonObject = JSONObject.parseObject(s);
                jsonObject.put("source", "flink");
                return jsonObject.toString();
            }
        });
        map.print();
//        KafkaSink.<JSONObject>builder().setBootstrapServers(kafka_server)
//
//        env.fromSource()

        KafkaSink<String> sink = KafkaSink.<String>builder()
                .setBootstrapServers(kafka_server)
                .setRecordSerializer(KafkaRecordSerializationSchema.builder()
                        .setTopic(sub_topic)
                        .setValueSerializationSchema(new SimpleStringSchema())
                        .build()
                )
                .setDeliveryGuarantee(DeliveryGuarantee.AT_LEAST_ONCE)
                .build();

        map.sinkTo(sink);

        //启动作业
        env.execute();

    }
}
